package com.main.crossdomaineventmanagement.event.config;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import com.main.crossdomaineventmanagement.event.model.EventMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.time.DateFormatUtils;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.protocol.admin.ConsumeStats;
import org.apache.rocketmq.remoting.protocol.admin.OffsetWrapper;
import org.apache.rocketmq.remoting.protocol.admin.TopicStatsTable;
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.*;

@Component
@EnableScheduling
@Slf4j
public class RocketMQUnconsumedMonitor {

    @Value("${rocketmq.name-server}")
    private String nameServer;
    private final ObjectMapper objectMapper = new ObjectMapper()
            .registerModule(new JavaTimeModule());  // 仅注册 JavaTime 模块


//    @Scheduled(cron = "0/30 * * * * ?")  // 每5秒执行
    public void checkUnconsumedDefaultMessages() {
        try {
            DefaultMQAdminExt admin = new DefaultMQAdminExt();
            admin.setNamesrvAddr(nameServer);
            admin.start();

            String topic = "DefaultEventsTopic";
            String consumerGroup = "DefaultConsumerGroup";

            // 获取消费进度
            ConsumeStats consumeStats = admin.examineConsumeStats(consumerGroup);
            TopicStatsTable topicStats = admin.examineTopicStats(topic);

            // 计算所有队列的未消费总量
            long totalUnconsumed = 0;
            for (MessageQueue mq : topicStats.getOffsetTable().keySet()) {

                OffsetWrapper offsetWrapper = consumeStats.getOffsetTable().getOrDefault(mq,
                        new OffsetWrapper());

                long brokerOffset = offsetWrapper.getBrokerOffset();
                long consumerOffset = offsetWrapper.getConsumerOffset();
                System.out.println(mq);
                System.out.println("getLastTimestamp:" +  DateFormatUtils.format(new Date(offsetWrapper.getLastTimestamp()), "yyyy-MM-dd HH:mm:ss"));
                System.out.println("brokerOffset:" +  brokerOffset);
                System.out.println("consumerOffset:" +  consumerOffset);
                totalUnconsumed += (brokerOffset - consumerOffset);
                if (brokerOffset > consumerOffset) {
                    pullUnconsumedMessages(topic, consumerGroup, mq, consumerOffset, brokerOffset);
                }
            }
            System.out.println("Topic " + topic + " 未消费总量: " + totalUnconsumed);

        } catch (MQClientException e) {
            log.error("MQ客户端初始化失败", e);
        } catch (Exception e) {
            log.error("查询未消费消息异常", e);
        }
    }

//    @Scheduled(cron = "0/30 * * * * ?")  // 每5秒执行
    public void checkUnconsumedTransactionMessages() {
        try {
            DefaultMQAdminExt admin = new DefaultMQAdminExt();
            admin.setNamesrvAddr(nameServer);
            admin.start();

            String topic = "TransactionEventsTopic";
            String consumerGroup = "TransactionConsumerGroup";

            // 获取消费进度
            ConsumeStats consumeStats = admin.examineConsumeStats(consumerGroup);
            TopicStatsTable topicStats = admin.examineTopicStats(topic);

            // 计算所有队列的未消费总量
            long totalUnconsumed = 0;
            for (MessageQueue mq : topicStats.getOffsetTable().keySet()) {

                OffsetWrapper offsetWrapper = consumeStats.getOffsetTable().getOrDefault(mq,
                        new OffsetWrapper());

                long brokerOffset = offsetWrapper.getBrokerOffset();
                long consumerOffset = offsetWrapper.getConsumerOffset();
                System.out.println(mq);
                System.out.println("getLastTimestamp:" +  DateFormatUtils.format(new Date(offsetWrapper.getLastTimestamp()), "yyyy-MM-dd HH:mm:ss"));
                System.out.println("brokerOffset:" +  brokerOffset);
                System.out.println("consumerOffset:" +  consumerOffset);
                totalUnconsumed += (brokerOffset - consumerOffset);
                if (brokerOffset > consumerOffset) {
                    pullUnconsumedMessages(topic, consumerGroup, mq, consumerOffset, brokerOffset);
                }
            }
            System.out.println("Topic " + topic + " 未消费总量: " + totalUnconsumed);

        } catch (MQClientException e) {
            log.error("MQ客户端初始化失败", e);
        } catch (Exception e) {
            log.error("查询未消费消息异常", e);
        }
    }

    private void pullUnconsumedMessages(String topic, String consumerGroup, MessageQueue mq,
                                        long consumerOffset, long brokerOffset) throws MQClientException {
        DefaultLitePullConsumer consumer = null;
        try {
            consumer = new DefaultLitePullConsumer(consumerGroup);
            consumer.setNamesrvAddr(nameServer); // 假设nameServer是类的成员变量
            consumer.setPullBatchSize(32); // 每次拉取的消息数量，可调整
            consumer.start();

            // 分配指定的MessageQueue并定位到consumerOffset
            consumer.assign(Collections.singletonList(mq));
            consumer.seek(mq, consumerOffset);

            long currentOffset = consumerOffset;
            while (currentOffset < brokerOffset) {
                List<MessageExt> messages = consumer.poll(1000);
                if (messages.isEmpty()) {
                    // 检查是否已到达brokerOffset
                    long committedOffset = consumer.committed(mq);
                    if (committedOffset >= brokerOffset) {
                        break;
                    }
                    continue;
                }

                for (MessageExt message : messages) {
                    // 处理消息：此处仅打印，可根据需求扩展
                    log.info("拉取到未消费消息: Topic={}, QueueId={}, Offset={}, MsgId={}, Tags={}",message.getTopic(),message.getQueueId(),message.getQueueOffset(),message.getMsgId(), message.getTags());
                    EventMessage event = objectMapper.readValue(message.getBody(), EventMessage.class);
                    log.info("未消费消息{}，原始内容: {}", message.getMsgId(), event.getPayload());
                    currentOffset = message.getQueueOffset() + 1; // 更新为下一条的offset
                }

                if (currentOffset >= brokerOffset) {
                    break;
                }
            }
        } catch (Exception e) {
            log.error("拉取未消费消息失败", e);
            throw new MQClientException("拉取消息异常", e);
        } finally {
            if (consumer != null) {
                consumer.shutdown();
            }
        }
    }

}